//package com.fanxl.flink.datastream.iterate;
//
//import org.apache.flink.api.common.functions.FilterFunction;
//import org.apache.flink.api.common.functions.MapFunction;
//import org.apache.flink.api.java.tuple.Tuple2;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.datastream.IterativeStream;
//import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
//import org.apache.flink.streaming.api.windowing.time.Time;
//import org.apache.flink.streaming.runtime.operators.util.AssignerWithPeriodicWatermarksAdapter;
//
///**
// * @author Dell
// * @Title:
// * @Description: 请写注释类
// * @date 2022/7/8
// */
//public class IterativeStreamTest {
//
//    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        test1(env);
//        env.execute("EventTimeWMApp");
//    }
//
//
//    public static void test1(StreamExecutionEnvironment env){
//        SingleOutputStreamOperator<String> lines = env.socketTextStream("localhost", 9527)
//                .assignTimestampsAndWatermarks(
//                        new AssignerWithPeriodicWatermarksAdapter.Strategy<>(
//                                new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(2)) {
//                                    @Override
//                                    public long extractTimestamp(String element) {
//                                        return Long.parseLong(element.split(",")[0]);
//                                    }
//                                }));
//
//        SingleOutputStreamOperator<Tuple2<String, Integer>> mapStream = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {
//            @Override
//            public Tuple2<String, Integer> map(String value) throws Exception {
//                String[] splits = value.split(",");
//                return Tuple2.of(splits[1].trim(), Integer.parseInt(splits[2].trim()));
//            }
//        });
//        SingleOutputStreamOperator<Tuple2<String, Integer>> filter = mapStream.filter(new FilterFunction<Tuple2<String, Integer>>() {
//            @Override
//            public boolean filter(Tuple2<String, Integer> value) throws Exception {
//                return 3 < value.f1 && value.f1 < 5;
//            }
//        });
//
//        SingleOutputStreamOperator<Tuple2<String, Integer>> filter1 = mapStream.filter(new FilterFunction<Tuple2<String, Integer>>() {
//            @Override
//            public boolean filter(Tuple2<String, Integer> value) throws Exception {
//                return 4 < value.f1 && value.f1 < 6;
//            }
//        });
//        SingleOutputStreamOperator<Tuple2<String, Integer>> filter2 = mapStream.filter(new FilterFunction<Tuple2<String, Integer>>() {
//            @Override
//            public boolean filter(Tuple2<String, Integer> value) throws Exception {
//                return 5 < value.f1 && value.f1 < 7;
//            }
//        });
//        IterativeStream<Tuple2<String, Integer>> iteration = filter.iterate();
//        iteration.filter(new FilterFunction<Tuple2<String, Integer>>() {
//            @Override
//            public boolean filter(Tuple2<String, Integer> value) throws Exception {
//                return value != null;
//            }
//        });
////        iteration.withFeedbackType()
//    }
//    public void test2(){
//        DataStream<Long> someIntegers = env.generateSequence(0, 1000);
//
//        IterativeStream<Long> iteration = someIntegers.iterate();
//
//        DataStream<Long> minusOne = iteration.map(new MapFunction<Long, Long>() {
//            @Override
//            public Long map(Long value) throws Exception {
//                return value - 1 ;
//            }
//        });
//
//        DataStream<Long> stillGreaterThanZero = minusOne.filter(new FilterFunction<Long>() {
//            @Override
//            public boolean filter(Long value) throws Exception {
//                return (value > 0);
//            }
//        });
//
//        iteration.closeWith(stillGreaterThanZero);
//
//        DataStream<Long> lessThanZero = minusOne.filter(new FilterFunction<Long>() {
//            @Override
//            public boolean filter(Long value) throws Exception {
//                return (value <= 0);
//            }
//        });
//    }
//}
